package com.test.custom.partition;

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;

import java.util.Map;
import java.util.Random;

/**
 * <dl>
 * <dt>CustomePartition</dt>
 * <dd>Description:自定义分区</dd>
 * <dd>Copyright: Copyright (C) 2006</dd>
 * <dd>CreateDate: 2019-3-7</dd>
 * </dl>
 *
 * @author Administrator
 */
public class CustomePartition implements Partitioner {
    /**
     *
     * @param topic 主题
     * @param key
     * @param keyBytes key的字节数组，如果对象是二进制流
     * @param value
     * @param valueBytes
     * @param cluster 集群信息
     * @return
     */
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        //拿到主题下有几个分区
        Integer partitionC = cluster.partitionCountForTopic(topic);
        if(null != key){
            if(key.toString().startsWith("1") || key.toString().startsWith("2") || key.toString().startsWith("3")
                    || key.toString().startsWith("4")){
                return 0;
            }else{
                return 1;
            }
        }
        return new Random().nextInt(partitionC);
    }

    public void close() {

    }

    public void configure(Map<String, ?> configs) {

    }
}
